package raft

// The file raftapi/raft.go defines the interface that raft must
// expose to servers (or the tester), but see comments below for each
// of these functions for more details.
//
// Make() creates a new raft peer that implements the raft interface.

import (
	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	"6.5840/labgob"
	"6.5840/labrpc"
	"6.5840/raftapi"
	"6.5840/tester1"
)

const (
	FOLLOWER = iota
	CANDIDATE
	LEADER
)

const (
	VOTED_NOBODY         = -1
	MIN_ELECTION_TIMEOUT = 150 //ms
	MAX_ELECTION_TIMEOUT = 450 //ms
	HEARTBEAT_INTERVAL   = 100 //ms
	TIGHT_LOOP_INTERVAL  = 10  //ms
	N_MAX_TRY            = 3
	N_MAX_CONCURRENT_AE  = 10
)

type rpcServiceBaseMsg struct {
	me      int
	n_peers int
}

// A Go object implementing a single Raft peer.
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *tester.Persister   // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (3A, 3B, 3C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	// persistent state
	CurrentTerm int
	VotedFor    int
	Log         []Entry

	commitIndex int
	lastApplied int

	lastConsistentIndex int

	nextIndex  []int
	matchIndex []int

	status          int32
	electionTimeout bool // 如果不为false,ticker就要开选举了。收到heartbeat就reset

	appendEntriesChan []chan int

	lastIncludedIndex int
	snapshot          []byte

	applyChannel chan raftapi.ApplyMsg

	startElectionChannel  chan int
	startHeartbeatChannel chan int
	startApplyChannel     chan int
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	// Your code here (3A).
	rf.mu.Lock()
	term = rf.CurrentTerm
	isleader = (rf.status == LEADER)
	rf.mu.Unlock()
	return term, isleader
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
// before you've implemented snapshots, you should pass nil as the
// second argument to persister.Save().
// after you've implemented snapshots, pass the current snapshot
// (or nil if there's not yet a snapshot).
func (rf *Raft) persist() {
	// Your code here (3C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// raftstate := w.Bytes()
	// rf.persister.Save(raftstate, nil)

	stateBuffer := new(bytes.Buffer)
	stateEncoder := labgob.NewEncoder(stateBuffer)
	stateEncoder.Encode(rf.CurrentTerm)
	stateEncoder.Encode(rf.VotedFor)
	stateEncoder.Encode(rf.lastIncludedIndex)
	stateEncoder.Encode(rf.Log) // log []Entry
	// snapshot好像可以直接放进去
	rf.persister.Save(stateBuffer.Bytes(), rf.snapshot)
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (3C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }

	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var currentTerm int
	var votedFor int
	var lastIncludedIndex int
	var log []Entry
	if d.Decode(&currentTerm) == nil &&
		d.Decode(&votedFor) == nil &&
		d.Decode(&lastIncludedIndex) == nil &&
		d.Decode(&log) == nil {
		rf.mu.Lock()
		rf.CurrentTerm = currentTerm
		rf.VotedFor = votedFor
		rf.lastIncludedIndex = lastIncludedIndex
		rf.Log = log
		rf.mu.Unlock()
	}
}

// how many bytes in Raft's persisted log?
func (rf *Raft) PersistBytes() int {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	return rf.persister.RaftStateSize()
}

// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	// Your code here (3B).

	rf.mu.Lock()
	isLeader = (rf.status == LEADER)
	if isLeader {
		rf.Log = append(rf.Log, Entry{Term: rf.CurrentTerm, Command: command})
		rf.nextIndex[rf.me] = rf.relative2Abs(len(rf.Log)) // 其实我觉得这个没必要
		rf.matchIndex[rf.me] = rf.nextIndex[rf.me] - 1
		index = rf.matchIndex[rf.me]
		term = rf.CurrentTerm
		// Debug(dLeader, "S%d startAppendEntriesService with status: %v", rf.me, rf)
		for i := range len(rf.peers) {
			if i != rf.me {
				go rf.appendEntriesTo(i)
			}
		}
		// go rf.startAppendEntriesService(&appendEntriesMsg{rpcServiceBaseMsg: rpcServiceBaseMsg{me: rf.me, n_peers: len(rf.peers)}})
		rf.persist()
	}
	rf.mu.Unlock()

	return index, term, isLeader
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) ticker() {
	for rf.killed() == false {

		// Your code here (3A)
		// Check if a leader election should be started.

		rf.mu.Lock()
		if rf.status != LEADER && rf.electionTimeout {
			rf.becomeCandidate()
		}
		rf.electionTimeout = true
		rf.mu.Unlock()

		// pause for a random amount of time between 50 and 350
		// milliseconds.
		// ms := 50 + (rand.Int63() % 300)
		// time.Sleep(time.Duration(ms) * time.Millisecond)
		ms := MIN_ELECTION_TIMEOUT + (rand.Int63() % (MAX_ELECTION_TIMEOUT - MIN_ELECTION_TIMEOUT))
		time.Sleep(time.Duration(ms) * time.Millisecond)
	}
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *tester.Persister, applyCh chan raftapi.ApplyMsg) raftapi.Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (3A, 3B, 3C).
	// Debug(dInfo, "S%d starts", rf.me)
	rf.CurrentTerm = 0
	rf.Log = []Entry{{Term: 0, Command: nil}}

	rf.commitIndex = 0 //! 检查一下是不是单增的
	rf.lastApplied = 0 //! 检查一下是不是单增的

	rf.nextIndex = make([]int, len(peers))
	rf.matchIndex = make([]int, len(peers))

	rf.becomeFollower()
	rf.electionTimeout = false

	rf.appendEntriesChan = make([]chan int, len(peers))
	for i := range rf.peers {
		rf.appendEntriesChan[i] = make(chan int, N_MAX_CONCURRENT_AE)
	}

	rf.applyChannel = applyCh

	rf.startElectionChannel = make(chan int)
	rf.startHeartbeatChannel = make(chan int)
	rf.startApplyChannel = make(chan int)

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	rf.snapshot = persister.ReadSnapshot()
	if rf.snapshot != nil {
		go rf.startApplyService(0)
	}

	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.electionService()
	go rf.heartbeatService()
	go rf.applyService()

	return rf
}
